package com.gitee.xhs.window.triggers;

import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.Trigger;
import org.apache.flink.streaming.api.windowing.triggers.TriggerResult;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.flink.streaming.api.windowing.triggers.TriggerResult.*;

/**
 * @author BigHead
 * @date 2022/7/26 16:43
 * @description CustomOnEventTimeTrigger
 */
public class CustomOnEventTimeTrigger extends Trigger<Transaction, TimeWindow> {

    private static final Logger logger = LoggerFactory.getLogger(CustomOnEventTimeTrigger.class);

    @Override
    public TriggerResult onElement(Transaction element, long timestamp, TimeWindow window, TriggerContext ctx) throws Exception {
        ctx.registerEventTimeTimer(element.getTimestamp() + Time.seconds(1).toMilliseconds());
        return CONTINUE;
    }

    @Override
    public TriggerResult onProcessingTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
        return CONTINUE;
    }

    @Override
    public TriggerResult onEventTime(long time, TimeWindow window, TriggerContext ctx) throws Exception {
        return FIRE;
    }

    @Override
    public void clear(TimeWindow window, TriggerContext ctx) throws Exception {
        logger.info("CustomOnEventTimeTrigger Clear" + window);
    }
}
